package GraghX

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.{SparkConf, SparkContext}

object graphX_demo2{

  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("Pregel API GraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 构建图
    val myVertices = sc.parallelize(Array((1L, 0), (2L, 0), (3L, 0), (4L, 0),
      (5L, 0)))
    val myEdges = sc.makeRDD(Array( Edge(1L,2L,2.5),
      Edge(2L,3L,3.6) , Edge(3L,4L,4.5),
      Edge(4L,5L,0.1),Edge(3L,5L,5.2)
    ))
    val myGraph = Graph(myVertices,myEdges)

    val sourceId: VertexId = 1L // The ultimate source
    // Initialize the graph such that all vertices except the root have distance infinity.
    val initialGraph = myGraph.mapVertices((id, _) =>
      if (id == sourceId) 0.0 else Double.PositiveInfinity)


    val sssp: Graph[Double, Double] = initialGraph.pregel(Double.PositiveInfinity)(
      (id, dist, newDist) => math.min(dist, newDist), // Vertex Program
      triplet => {  // Send Message
        if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
          Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
        } else {
          Iterator.empty
        }
      },
      (a, b) => math.min(a, b) // Merge Message
    )

    sssp.vertices.collect.foreach(println(_))
  }


}
